package org.zjt.spark.book

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.zjt.spark.book.KDistance.sc

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
  * DESC    
  *
  * @author
  * @create 2017-06-27 下午4:34
  **/
object KClassify extends App {
  var sparkConf = new SparkConf().setMaster("local[2]").setAppName("WeatherSort")
  val sc = new SparkContext(sparkConf)
  val k: Broadcast[Int] = sc.broadcast[Int](2)
  val rddCenter = sc.textFile("/Users/zhangjuntao/IdeaProjects/myproject/hw-bigdata/scala-demo/src/main/resource/k-center.log").map {
    line => {
      val array: Array[String] = line.split("   ")
      val className = array.head
      val value = array.tail.reduce((sum, b) => sum + b)
      (value, className)
    }
  }

  val rddData = sc.textFile("/Users/zhangjuntao/IdeaProjects/myproject/hw-bigdata/scala-demo/src/main/resource/K.log").map {
    line => {
      //zhang2	92.37583667639797	58.09416642856334
      val array: Array[String] = line.split("\t")
      val k = "[%s,%s]".format(array(1), array(2))
      (array(0), k)
    }
  }


  // rdd生成笛卡尔积    100 * 4个数据，相当于union
  val rdd = rddData.cartesian(rddCenter).persist().groupByKey().map {
    line => {
      val values: scala.Iterable[(String, String)] = line._2
      val location = line._1._2
      val kNears = KHandler.findNearK(location, values, k.value)
      val major = KHandler.majorNear(kNears)
      (line._1, major)
    }
  }

  println(rdd.count())
  rdd.collect().foreach(a => println(a._1 + ":" + (a._2.mkString(","))))

  sc.stop()
}


object KHandler {

  /**
    *
    * @param location [272.95004663694726,229.63001281966578]
    * @param values   values
    * @param k        得到K个近邻
    * @return
    */
  def findNearK(location: String, values: scala.Iterable[(String, String)], k: Int): Array[(String, String)] = {
    var sort = ArrayBuffer[((String, String), Double)]()
    // [272.95004663694726,229.63001281966578]
    val array = location.replace("[", "").replace("]", "").split(",")
    val locationX = array(0).toDouble
    val locationY = array(1).toDouble
    for (value <- values) {
      //[121.40984699542561,87.02053927940713]
      val vector = value._1.replace("[", "").replace("]", "").split(",")
      val x = vector(0).toDouble
      val y = vector(1).toDouble
      val dis = Math.sqrt(Math.pow(x - locationX, 2) + Math.pow(y - locationY, 2))
      val element = (value, dis)
      sort += element
    }
    sort.sortBy(_._2).take(k).map(_._1).toArray
  }


  def majorNear(values: Array[(String, String)]): Array[(String, String)] = {
    var map = new mutable.HashMap[String, Int]
    for (value <- values) {
      val cnt = map.getOrElse(value._2, 0) + 1
      map += (value._2 -> cnt)
    }
    val maxCnt: Int = map.values.max

    var max: String = null
    for (key <- map.keySet) {
      val v = map.get(key) match {
        case Some(a) => a
        case None => 0
      }
      if (v >= maxCnt) {
        max = key
      }
    }
    values.filter( a => a._2.equals(max))
  }

}
